package com.tlb.springboot.starter.rocketmq.annotation;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.UUID;

import com.tlb.springboot.starter.rocketmq.config.RocketMqProperties;
import com.tlb.springboot.starter.rocketmq.impl.DefaultMessageListenerConcurrently;
import com.tlb.springboot.starter.rocketmq.impl.DefaultMessageListenerOrderly;
import com.tlb.springboot.starter.rocketmq.util.AliAddressUtil;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;

import com.aliyun.openservices.ons.api.impl.rocketmq.OnsClientRPCHook;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.MessageSelector;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.exception.MQClientException;

import lombok.extern.slf4j.Slf4j;

/**
 * @author yuaneg 2018/8/30
 */
@Slf4j
public class RocketMQListenBeanPostProcessor implements BeanPostProcessor {

    private RocketMqProperties rocketMqProperties;

    private Environment environment;

    public RocketMQListenBeanPostProcessor(RocketMqProperties rocketMqProperties, Environment environment) {
        this.rocketMqProperties = rocketMqProperties;
        this.environment = environment;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        Map<Method, RocketMQListen> annotatedMethods = MethodIntrospector.selectMethods(targetClass, this::inspect);
        if (!annotatedMethods.isEmpty()) {
            for (Method method : annotatedMethods.keySet()) {
                RocketMQListen rocketMQListen = annotatedMethods.get(method);
                DefaultMQPushConsumer defaultMQPushConsumer;
                if (null == rocketMqProperties.getSessionCredentials()) {
                    defaultMQPushConsumer = new DefaultMQPushConsumer();
                    defaultMQPushConsumer.setNamesrvAddr(rocketMqProperties.getNameServer());
                } else {
                    defaultMQPushConsumer = new DefaultMQPushConsumer(
                        AliAddressUtil.getNamespace(rocketMqProperties.getNameServer()),
                        replace(rocketMQListen.consumerGroup()),
                        new OnsClientRPCHook(rocketMqProperties.getSessionCredentials()));
                    defaultMQPushConsumer
                        .setNamesrvAddr(AliAddressUtil.fetchNSAddr(rocketMqProperties.getNameServer()));
                }
                defaultMQPushConsumer.setVipChannelEnabled(false);
                defaultMQPushConsumer.setConsumerGroup(replace(rocketMQListen.consumerGroup()));
                defaultMQPushConsumer.setConsumeTimeout(rocketMQListen.consumeTimeout());
                defaultMQPushConsumer.setMaxReconsumeTimes(rocketMQListen.maxReconsumeTimes());
                defaultMQPushConsumer.setConsumeThreadMax(rocketMQListen.consumeThreadMax());
                try {
                    switch (rocketMQListen.selectorType()) {
                        case TAG:
                            defaultMQPushConsumer.subscribe(replace(rocketMQListen.topic()), rocketMQListen.tags());
                            break;
                        case SQL92:
                            defaultMQPushConsumer.subscribe(replace(rocketMQListen.topic()), MessageSelector.bySql(rocketMQListen.tags()));
                            break;
                        default:
                            throw new IllegalArgumentException("Property 'selectorType' was wrong.");
                    }

                    switch (rocketMQListen.consumeMode()) {
                        case ORDERLY:
                            defaultMQPushConsumer.setMessageListener(new DefaultMessageListenerOrderly(bean, method, rocketMQListen));
                            break;
                        case CONCURRENTLY:
                            defaultMQPushConsumer.setMessageListener(new DefaultMessageListenerConcurrently(bean, method, rocketMQListen));
                            break;
                        default:
                            throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
                    }
                } catch (MQClientException e) {
                    log.error("rocketMQ consumer异常:" + rocketMQListen.topic(), e);
                }
                defaultMQPushConsumer.setInstanceName(UUID.randomUUID().toString());
                try {
                    defaultMQPushConsumer.start();
                } catch (MQClientException e) {
                    log.error("rocketMQ consumer启动失败topic:" + rocketMQListen.topic(), e);
                }
            }
        }
        return bean;
    }

    private RocketMQListen inspect(Method method) {
        return AnnotationUtils.findAnnotation(method, RocketMQListen.class);
    }

    private String replace(String value) {
        if (value.contains("${")) {
            value = value.replace("}", "");
            value = value.replace("${", "");
            return environment.getProperty(value);
        }
        return value;
    }

}
